/*******************************************************************************
 * Copyright (c) 2012, 2017 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *   http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial contribution
 *******************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <unistd.h>
#include "MQTTClient.h"
#include <sys/msg.h>



#define ADDRESS "tcp://172.16.74.132:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC_send "/app/control/up"
#define TOPIC_recv "/app/control/down"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L
#define MAX_TEXT 512//队列文本大小
#define MSG_KEY 1234//队列密钥
#define SHM_KEY 1234//内存共享密钥
#define SEM_KEY 5678//内存信号锁密钥
#define SHM_SIZE 512//共享内存文本大小

struct shmseg {
    int written;
    char text[SHM_SIZE];
};//内存文本结构体

//// 信号量操作：P操作（等待）
void sem_wait_nonblock(int semid) {
    struct sembuf sb = {0, -1, IPC_NOWAIT};
    if (semop(semid, &sb, 1) == -1) {
        // 如果没有资源，直接返回，不阻塞
        return;
    }
}


//// 信号量操作：V操作（释放）
void sem_signal(int semid) {
    struct sembuf sb = {0, 1, 0};
    semop(semid, &sb, 1);
}


struct msgbuf {
    long mtype;
    char mtext[MAX_TEXT];
};//队列文本结构体

volatile MQTTClient_deliveryToken deliveredtoken;
// 全局变量
// MQTTClient client;
pthread_mutex_t client_mutex = PTHREAD_MUTEX_INITIALIZER;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
int running = 1;

void delivered(void *context, MQTTClient_deliveryToken dt) // 回调函数查看消息是否被服务器接收到
{
    //printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;
    for (i = 0; i < message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
} // 回调函数处理接收到的数据

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
//接收内存
void *recv_neicun_thread(void*arg)
{
    int shmid, semid;
    struct shmseg *shm;

    shmid = shmget(SHM_KEY, sizeof(struct shmseg), 0666);
    if (shmid == -1) {
        perror("shmget failed");
        exit(1);
    }

    semid = semget(SEM_KEY, 1, 0666);
    if (semid == -1) {
        perror("semget failed");
        exit(1);
    }

    shm = (struct shmseg *) shmat(shmid, NULL, 0);
    if (shm == (void *) -1) {
        perror("shmat failed");
        exit(1);
    }

    while (running) {
    sem_wait_nonblock(semid);
    if (shm->written == 1) {
        printf("Received: %s", shm->text);
        shm->written = 0;
    }
    sem_signal(semid);
    usleep(100000); // 更频繁轮询
}


    
}
//发送队列信息函数
void *send_duilie_thread(void *arg)
{
    int msgid;
    struct msgbuf msg;

    msgid = msgget(MSG_KEY, 0666 | IPC_CREAT);
    if (msgid == -1) {
        perror("msgget failed");
        exit(EXIT_FAILURE);
    }

    printf("Enter messages to send (type 'exit' to quit):\n");
    while (running) {
        printf("> ");
        if (fgets(msg.mtext, MAX_TEXT, stdin) == NULL) {
            perror("fgets failed");
            break;
        }

        if (strncmp(msg.mtext, "exit", 4) == 0) {
            running = 0;  // 通知其他线程退出
            break;
        }

        msg.mtype = 1;
        if (msgsnd(msgid, &msg, strlen(msg.mtext) + 1, 0) == -1) {
            perror("msgsnd failed");
            break;
        }
    }

    pthread_exit(NULL);
}

// 接收线程函数（主动拉取消息）mqtt
void *receive_thread(void *arg)
{

    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n",
           TOPIC_recv, CLIENTID, QOS);

    MQTTClient_message *msg = NULL;

    char *topic = NULL;
    while (running)
    {

        pthread_mutex_lock(&client_mutex);
        int rc = MQTTClient_receive(client, &topic, NULL, &msg, 1000); // 阻塞5秒
        pthread_mutex_unlock(&client_mutex);

        if (rc == MQTTCLIENT_SUCCESS && msg != NULL)
        {
            printf("Received on [%s]: %.*s\n",
                   topic, msg->payloadlen, (char *)msg->payload);

            // 必须释放内存！
            MQTTClient_free(topic);
            MQTTClient_freeMessage(&msg);
        }
        // else
        // {
        //     printf("No message received (rc=%d)\n", rc);
        // }
    }

    // return rc;
}

// 发送线程函数
void *send_thread(void *arg)
{
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    pubmsg.payload = PAYLOAD;
    pubmsg.payloadlen = (int)strlen(PAYLOAD);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    int rc=0;

    while (running)
    {

        pthread_mutex_lock(&client_mutex);
        rc=MQTTClient_publishMessage(client, TOPIC_send, &pubmsg, &token);
        if (rc != MQTTCLIENT_SUCCESS) {
        printf("Publish failed: %d\n", rc);
            }
        // printf("Waiting for up to %d seconds for publication of %s\n"
        //        "on topic %s for client with ClientID: %s\n",
        //        (int)(TIMEOUT / 1000), PAYLOAD, TOPIC_send, CLIENTID);
        // // rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
        // printf("Message with delivery token %d delivered\n", token);
        pthread_mutex_unlock(&client_mutex);
    }

    // return rc;
}

int main(int argc, char *argv[])
{
    int rc = 0;
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 订阅线程//订阅的是mqtt主题
    pthread_mutex_lock(&client_mutex);
    MQTTClient_subscribe(client, TOPIC_recv, QOS);
    pthread_mutex_unlock(&client_mutex);

    // 创建线程
    pthread_t recv_tid, send_tid,send_duilie,recv_neicun;
    pthread_create(&recv_tid, NULL, receive_thread, NULL);
    pthread_create(&send_tid, NULL, send_thread, NULL);
    pthread_create(&send_duilie, NULL, send_duilie_thread, NULL);
    pthread_create(&recv_neicun, NULL, recv_neicun_thread, NULL);
    // 主线程控制退出
    // 不再 getchar()，由输入 exit 控制退出
while (running) {
    sleep(1);
}
  
    
    
    pthread_join(send_tid, NULL);
    pthread_join(recv_tid, NULL);
    pthread_join(send_duilie,NULL);
    pthread_join(recv_neicun,NULL);
    MQTTClient_unsubscribe(client, TOPIC_recv);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    pthread_mutex_destroy(&client_mutex);
    return 0;
}
